package com.too.demo.message;

import com.too.demo.util.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;

import java.util.Map;

/**
 * mq消费者
 *
 * @author ouyf
 * @date 2024/9/5 13:31
 * <p>
 * Copyright (c) Siemens Mobility TBDS 2024, All Rights Reserved, Confidential
 */
@Slf4j
public class RedisMessageConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private RedisStreamUtil redisStreamUtil;


    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class);
        // stream的key值
        String streamKey = message.getStream();
        //消息ID
        RecordId recordId = message.getId();
        //消息内容
        Map<String, String> msg = message.getValue();

        System.out.println("streamKey=" + streamKey + ",recordId=" + recordId + ",msg" + msg);
        //TODO 处理逻辑

        String group = "VS_GROUP";

        //逻辑处理完成后，ack消息，删除消息，group为消费组名称
        redisStreamUtil.ack(streamKey,group,recordId.getValue());
//        redisStreamUtil.del(streamKey,recordId.getValue());

        System.out.println("successful consumer msgId=" + recordId);
    }
}
